package com.redis.queue;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.Record;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

/**
 * redis stream 队列 公共方法
 */
@Component
@Slf4j
public class RedisQueueUtils {

    @Autowired
    private StringRedisTemplate redisTemplate;

    /**
     * 创建消费组
     *     1.创建消费流
     *     2.创建消费组
     * create time: 2022/5/11 16:45
     * @param:
     * @return java.lang.String
     */
    public void createGroup(String key, String group){
        if (!redisTemplate.hasKey(key)){
            redisTemplate.opsForStream().add(key,Collections.singletonMap("", ""));
            log.info("Redis stream 队列不存在 已初始化stream {} success",key);
        }
        //创建消费者组
        try {
            redisTemplate.opsForStream().createGroup(key, group);
        } catch (Exception e) {
            log.info("创建消费组失败,消费者组 {} 已存在",group);
        }

    }

    /**
     * description: 添加Map消息
     * create time: 2022/5/11 16:28
     * @param: key
     * @param: value
     * @return
     */
    public String addMap(String key, Object object){
        Map<String, String> map = new HashMap<>();
        map.put("object", JSON.toJSONString(object));
        return redisTemplate.opsForStream().add(key, map).getValue();
    }

    /**
     * description: 添加Record消息
     * create time: 2022/5/11 16:30
     * @param: record
     * @return
     */
    public String addRecord(Record<String, Object> record){
        return redisTemplate.opsForStream().add(record).getValue();
    }

    /**
     * description: 确认消费
     * create time: 2022/5/19 11:21
     * @param: key
     * @param: group
     * @param: recordIds
     * @return java.lang.Long
     */
    public Long ack(String key, String group, String... recordIds){
        return redisTemplate.opsForStream().acknowledge(key, group, recordIds);
    }

    /**
     * description: 删除消息。当一个节点的所有消息都被删除，那么该节点会自动销毁
     * create time: 2022/7/18 15:33
     * @param: key
     * @param: recordIds
     * @return java.lang.Long
     */
    public Long del(String key, String... recordIds){
        return redisTemplate.opsForStream().delete(key, recordIds);
    }

}
